package com.partsoft.dits.mq;

import java.io.IOException;
import java.util.concurrent.Executor;

import org.apache.commons.pool.KeyedObjectPool;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.ibm.mq.MQC;
import com.ibm.mq.MQException;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueueManager;
import com.partsoft.dits.EdgeException;
import com.partsoft.dits.log.Log;

public class MQMessageTransmitter implements MessageTransmitter, InitializingBean {

	private String name;

	private KeyedObjectPool qmPool = null;

	private MQPutMessageOptions pmo = new MQPutMessageOptions();

	private Executor taskExecutor = null;

	public void setQmPool(KeyedObjectPool qmPool) {
		this.qmPool = qmPool;
	}

	public KeyedObjectPool getQmPool() {
		return qmPool;
	}

	public MQMessageTransmitter() {
		pmo.options = MQC.MQPMO_NONE;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public void sendMessage(String connName, String destName, byte[] msg) {
		try {
			doPutMessage(connName, destName, buildMessage(msg));
		} catch (IOException e) {
			Log.error(String.format("打包消息时候出错:%s", e.getMessage()), e);
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	public void sendMessage(String connName, String destName, String msg) {
		try {
			doPutMessage(connName, destName, buildMessage(msg));
		} catch (IOException e) {
			Log.error(String.format("打包消息时候出错:%s", e.getMessage()), e);
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	private void doPutMessage(String connName, String destName, MQMessage msg) {
		PutMessageRunner runner = new PutMessageRunner(this, connName, destName, msg);
		if (taskExecutor != null) {
			taskExecutor.execute(runner);
		} else {
			runner.run();
		}
	}

	private MQMessage buildMessage(String msg) throws IOException {
		MQMessage mqMess = new MQMessage();
		mqMess.clearMessage();
		mqMess.correlationId = MQC.MQCI_NONE;
		mqMess.messageId = MQC.MQMI_NONE;
		mqMess.write(msg.getBytes("utf-8"));
		return mqMess;
	}

	private MQMessage buildMessage(byte[] msg) throws IOException {
		MQMessage mqMess = new MQMessage();
		mqMess.clearMessage();
		mqMess.correlationId = MQC.MQCI_NONE;
		mqMess.messageId = MQC.MQMI_NONE;
		mqMess.write(msg);
		return mqMess;
	}

	public MQPutMessageOptions getMQPutMessageOptions() {
		return pmo;
	}

	private static class PutMessageRunner implements Runnable {

		final MQMessageTransmitter msService;

		final String connName;

		final String destName;

		final MQMessage msg;

		PutMessageRunner(MQMessageTransmitter msService, String connName, String destName, MQMessage msg) {
			this.msService = msService;
			this.connName = connName;
			this.destName = destName;
			this.msg = msg;
		}

		public MQMessageTransmitter getMQMessageTransmitter() {
			return msService;
		}

		public String getConnectName() {
			return connName;
		}

		public String getDestQueueName() {
			return destName;
		}

		public MQMessage getSendMessage() {
			return msg;
		}

		public void run() {
			begin();
			doSend();
			end();
		}

		public MQPutMessageOptions getMQPutMessageOptions() {
			return getMQMessageTransmitter().getMQPutMessageOptions();
		}

		public KeyedObjectPool getQmPool() {
			return getMQMessageTransmitter().getQmPool();
		}

		private void doSend() {
			MQQueueManager qmManager = null;
			String cName = getConnectName();
			String dName = getDestQueueName();
			MQMessage sendMessage = getSendMessage();
			MQPutMessageOptions pmo = getMQPutMessageOptions();
			int reSendWait = 1;
			boolean resend = true;
			while (resend) {
				try {
					qmManager = (MQQueueManager) getQmPool().borrowObject(cName);
				} catch (Throwable e) {
					Log.warn(new EdgeException(
							StringUtils.hasText(getMQMessageTransmitter().getName()) ? new MQPoolDescriptor(
									getMQMessageTransmitter().getName()) : null, String.format(
									"打开队列管理器(%s)时出错:%s，等待%d秒后准备重发。", cName, e.getMessage(), reSendWait), e));
					try {
						Thread.currentThread().wait(reSendWait * 1000);
					} catch (InterruptedException e1) {
					}
					continue;
				}
				boolean supportComit = true;
				try {
					qmManager.begin();
				} catch (MQException e) {
					if (e.completionCode == 2 && e.reasonCode == 2012) {
						supportComit = false;
					} else {
						Log.warn(new EdgeException(
								StringUtils.hasText(getMQMessageTransmitter().getName()) ? new MQPoolDescriptor(
										getMQMessageTransmitter().getName()) : null,
								String.format("发送消息至(QM:%s, Q:%s)时候出错:%s，等待%d秒后重发。", cName, dName, e.getMessage(),
										reSendWait), e));
						try {
							Thread.currentThread().wait(reSendWait * 1000);
						} catch (InterruptedException e1) {
						}
						continue;
					}
				}
				try {
					qmManager.put(dName, sendMessage, pmo);
					if (supportComit)
						qmManager.commit();
					resend = false;
				} catch (MQException e) {
					try {
						if (supportComit)
							qmManager.backout();
					} catch (MQException e1) {
					}
					Log.warn(new EdgeException(
							StringUtils.hasText(getMQMessageTransmitter().getName()) ? new MQPoolDescriptor(
									getMQMessageTransmitter().getName()) : null, String.format(
									"发送消息至(QM:%s, Q:%s)时候出错:%s，等待%d秒后重发。", cName, dName, e.getMessage(), reSendWait), e));
					try {
						Thread.currentThread().wait(reSendWait * 1000);
					} catch (InterruptedException e1) {
					}
				} finally {
					try {
						qmManager.disconnect();
					} catch (MQException e) {
					}
					try {
						qmManager.close();
					} catch (MQException e) {
					}
					try {
						getQmPool().returnObject(cName, qmManager);
					} catch (Exception e) {
					}
				}
			}
		}

		private void end() {
		}

		private void begin() {
		}
	}

	public Executor getTaskExecutor() {
		return taskExecutor;
	}

	public void setTaskExecutor(Executor taskExecutor) {
		this.taskExecutor = taskExecutor;
	}

	public void afterPropertiesSet() throws Exception {
		Assert.notNull(getQmPool());
	}

	public static void main(String[] args) {
		System.out.println(String.format("test:%d", 1));
	}

}
